package se.sics.drsm.components;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import se.sics.drsm.fd.events.PeerFailureSuspicion;
import se.sics.drsm.fd.events.StartProbingPeer;
import se.sics.drsm.fd.ports.FailureDetector;
import se.sics.drsm.inits.DRSMNodePSInit;
import se.sics.drsm.messages.JoinGroup;
import se.sics.drsm.messages.JoinGroupResponse;
import se.sics.drsm.messages.UpdateConfiguration;
import se.sics.drsm.model.DRSMAddress;
import se.sics.drsm.model.StateMachineConfiguration;
import se.sics.drsm.model.SuccessorList;
import se.sics.drsm.overlay_address.key.NumericRingKey;
import se.sics.drsm.ports.drsm.DRSMPeriodicStabilizationPort;
import se.sics.drsm.ports.drsm.DRSMRouter;
import se.sics.drsm.ports.drsm.events.*;
import se.sics.drsm.ports.statemachine.StateMachinePort;
import se.sics.drsm.ports.statemachine.events.AddPeer;
import se.sics.drsm.ports.statemachine.events.UpdatePeers;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.address.Address;
import se.sics.kompics.network.Network;
import se.sics.kompics.p2p.fd.StopProbingPeer;
import se.sics.kompics.p2p.fd.SuspicionStatus;
import se.sics.kompics.p2p.overlay.chord.LeaveRing;
import se.sics.kompics.timer.SchedulePeriodicTimeout;
import se.sics.kompics.timer.Timer;

import java.math.BigInteger;
import java.util.*;

/**
 * The <code>DRSMNodePS</code> class.
 *
 * @author Shadid Chowdhury <msrch@kth.se>
 */
public final class DRSMNodePS extends ComponentDefinition {

    Negative<DRSMPeriodicStabilizationPort> ps = negative(DRSMPeriodicStabilizationPort.class);
    Positive<DRSMRouter> router = positive(DRSMRouter.class);
    Positive<StateMachinePort> stateMachinePort = positive(StateMachinePort.class);
    Positive<Network> network = positive(Network.class);
    Positive<Timer> timer = positive(Timer.class);
    Positive<FailureDetector> epfd = positive(FailureDetector.class);

    private Logger logger;

    private long stabilizationPeriod;
    private BigInteger ringSize;
    private DRSMAddress self;
    private List<DRSMAddress> peers, successors, predecessors;
    private SuccessorList successorList;

    private HashMap<Address, DRSMAddress> neighborPeers;
    private HashMap<Address, Integer> neighborCounters;
    private HashMap<Address, UUID> fdRequests;

    // This group should be in the config
    private boolean isLeader; // TODO: This should be in the configuration
    private DRSMAddress myLeader;

    public DRSMNodePS() {
        neighborPeers = new HashMap<Address, DRSMAddress>();
        neighborCounters = new HashMap<Address, Integer>();
        fdRequests = new HashMap<Address, UUID>();

        subscribe(handleInit, control);

        subscribe(handleCreateRing, ps);
        subscribe(handleJoinSuccessorGroup, ps);
        subscribe(handleJoinGroup, network);
        subscribe(handleJoinGroupResponse, network);
        subscribe(handleLeaveRing, ps);

        subscribe(handleUpdatePeers, stateMachinePort);

        subscribe(handleDRSMLookupResponse, router);
        subscribe(handleGetPredecessorRequest, network);
        subscribe(handleGetPredecessorResponse, network);
        subscribe(handleGetSuccessorListRequest, network);
        subscribe(handleGetSuccessorListResponse, network);
        subscribe(handleLeaveNotification, network);
        subscribe(handleNotify, network);
        subscribe(handleStabilizeTimeout, timer);
        subscribe(handlePeerFailureSuspicion, epfd);
    }

    Handler<DRSMNodePSInit> handleInit = new Handler<DRSMNodePSInit>() {
        public void handle(DRSMNodePSInit init) {
            stabilizationPeriod = init.getStabilizationPeriod();
            int log2RingSize = init.getLog2RingSize();
            ringSize = new BigInteger("2").pow(log2RingSize);

            int successorListLength = init.getSuccessorListLength();

            self = init.getSelf();

            successorList = new SuccessorList(successorListLength, self,
                    ringSize);

            logger = LoggerFactory.getLogger(getClass().getName() + "@"
                    + self.getKey().getId());
        }
    };

    private Handler<DRSMCreateRing> handleCreateRing = new Handler<DRSMCreateRing>() {
        public void handle(DRSMCreateRing event) {
            logger.debug("CREATE");

            isLeader = true;
            predecessors = null;
            successors = null;
            successorList.setSuccessor(self); // TODO: Where do we use self as
            // successor

            // trigger newSuccessor
            //DRSMNewSuccessors newSuccessor = new DRSMNewSuccessors(self,
            //        successorList.getSuccessorListView());
            //trigger(newSuccessor, ps);

            // trigger newPredecessor
            //DRSMNewPredecessors newPredecessor = new DRSMNewPredecessors(self,
            //        predecessors);
            //trigger(newPredecessor, ps);

            List<DRSMAddress> peerList = new ArrayList<DRSMAddress>();
            peerList.add(self);

            trigger(new UpdateConfiguration(self, self, new StateMachineConfiguration(self, peerList, null, null)), network);


            logger.info("Join Completed");

            // trigger JoinRingCompleted
            JoinRingCompleted joinRingCompleted = new JoinRingCompleted(self);
            trigger(joinRingCompleted, ps);

            // set the stabilization timer
            SchedulePeriodicTimeout spt = new SchedulePeriodicTimeout(
                    stabilizationPeriod, stabilizationPeriod);
            spt.setTimeoutEvent(new StabilizeTimeout(spt));
            trigger(spt, timer);
        }
    };

    private Handler<DRSMJoinSuccessorGroup> handleJoinSuccessorGroup = new Handler<DRSMJoinSuccessorGroup>() {
        public void handle(DRSMJoinSuccessorGroup event) {
            DRSMAddress insider = event.getChordInsiders().poll();
            logger.debug("handling DRSMJoinSuccessorGroup self {} ",
                    event.getSelf());
            logger.debug("JOIN through " + insider);
            // predecessors = null;

            DRSMLookupRequest lookupRequest = new DRSMLookupRequest(
                    self.getKey(), event.getChordInsiders(), insider, true);

            trigger(lookupRequest, router);

            // trigger newPredecessor
            // DRSMNewPredecessors newPredecessor = new
            // DRSMNewPredecessors(self,
            // predecessors);
            // trigger(newPredecessor, ps);
        }
    };

    @SuppressWarnings("unchecked")
    private Handler<DRSMLookupResponse> handleDRSMLookupResponse = new Handler<DRSMLookupResponse>() {
        public void handle(DRSMLookupResponse event) {
            // TODO: For failure I think we can consider re-trying to another
            // peer in the group
            if (event.getStatus() == DRSMLookupResponse.DRSMLookupStatus.FAILURE) {
                logger.debug("CHORD_LOOKUP_FAILED");

                LinkedList<DRSMAddress> insiders = (LinkedList<DRSMAddress>) event
                        .getAttachment();
                DRSMAddress nextInsider = insiders.poll();

                // retry join
                DRSMLookupRequest request = new DRSMLookupRequest(
                        self.getKey(), insiders, nextInsider,
                        event.isJoinRequest());

                // cycle through other joining peers
                trigger(request, router);
                return;
            }

            logger.debug("CHORD_LOOKUP_RESP R({})={} ", event.getKey(),
                    event.getResponsible());

            if (event.isJoinRequest()) {
                logger.debug("Sending joinGroup request to {} ", event.getKey());
                updateMyKey(event.getKey());
                logger.debug("\n responsible peers {}", event.getResponsible());
                DRSMAddress myLeader = event.getResponsible().get(0); // get the
                // config
                // leader
                // instead
                logger.info("Sending joinGroup request to leader {} ", myLeader);
                trigger(new JoinGroup(self, myLeader), network);
            }
            // Here we notify our peers about my presence and also update
            // my router about my new peers

            NewPeers peerList = new NewPeers(event.getResponsible());
            // Have to add myself?

            // we got the real successor
            // FIXME: successor list should have closest neighours. Why are we
            // even setting any random node to be successors?
            successorList.setSuccessor(event.getResponsible().get(0));

            neighborAdded(successorList.getSuccessor());

            // TODO: all information passing should be centralized
            //GetSuccessorListRequest request = new GetSuccessorListRequest(
            //        RequestState.JOIN, self, successorList.getSuccessor());
            //trigger(request, network);

            // trigger newSuccessor
            // DRSMNewSuccessors newSuccessor = new DRSMNewSuccessors(self,
            //       successorList.getSuccessorListView());
            //trigger(newSuccessor, ps);

            logger.info("Join Completed");

            // trigger JoinRingCompleted
            JoinRingCompleted joinRingCompleted = new JoinRingCompleted(self);
            trigger(joinRingCompleted, ps);

            // set the stabilization timer                                              ´
            SchedulePeriodicTimeout spt = new SchedulePeriodicTimeout(
                    stabilizationPeriod, stabilizationPeriod);
            spt.setTimeoutEvent(new StabilizeTimeout(spt));
            trigger(spt, timer);
        }
    };

    private Handler<JoinGroup> handleJoinGroup = new Handler<JoinGroup>() {
        public void handle(JoinGroup event) {
            logger.debug("handling JoinGroup self {} ", event.getSource());

            if (isLeader) {
                trigger(new AddPeer(self, event.getDRSMSource()), stateMachinePort);
                //TODO joinResponse should be sent when re-configuration is done
                JoinGroupResponse jgroupResponse = new JoinGroupResponse(self,
                        event.getDRSMSource(), true, self, successors, predecessors);
                trigger(jgroupResponse, network);
            } else {

                // TODO: may be here we dont do any thing here rather we decide in state machine
            }


        }
    };

    private Handler<JoinGroupResponse> handleJoinGroupResponse = new Handler<JoinGroupResponse>() {
        public void handle(JoinGroupResponse event) {
            logger.debug("handling JoinGroupResponse received from self {} ", event.getSource());

            if (event.isSuccessful()) {
                logger.debug("Joining group completed ");
            } else {
                logger.warn("Joining group request failed {} ", self);
            }
        }
    };

    Handler<LeaveRing> handleLeaveRing = new Handler<LeaveRing>() {
        public void handle(LeaveRing event) {
            // FIXME
            // if (self == null)
            // return;
            //
            // if (predecessors != null) {
            // trigger(new LeaveNotification(self, predecessors, successorList,
            // predecessors), network);
            //
            // }
            // if (successorList != null && successorList.getSuccessor() !=
            // null) {
            // trigger(new LeaveNotification(self, predecessors, successorList,
            // successorList.getSuccessor()), network);
            // }
            // trigger(new LeaveRingCompleted(self), ps);
            //
            // // discard all state
            // // self = predecessors = null;
            // // successorList = null;
            // // TODO discard fdRequests and neighborPeers
        }
    };

    Handler<LeaveNotification> handleLeaveNotification = new Handler<LeaveNotification>() {
        public void handle(LeaveNotification event) {
            // FIXME
            // DRSMAddress leavingPeer = event.getFromPeer();
            //
            // if (leavingPeer.equals(predecessors)) {
            // // predecessors left
            // predecessors = event.getPredecessor();
            // // trigger newPredecessor
            // DRSMNewPredecessor newPredecessor = new DRSMNewPredecessor(self,
            // predecessors);
            // trigger(newPredecessor, ps);
            // }
            // if (successorList.getSuccessor().equals(leavingPeer)) {
            // // successor left
            // successorList.successorFailed(leavingPeer);
            // // trigger newSuccessor
            // NewSuccessorList newSuccessor = new NewSuccessorList(self,
            // successorList.getSuccessorListView());
            // trigger(newSuccessor, ps);
            // }
            // neighborRemoved(leavingPeer);
        }
    };

    private Handler<UpdatePeers> handleUpdatePeers = new Handler<UpdatePeers>() {
        public void handle(UpdatePeers event) {
            logger.debug("handleUpdatePeers {}", event.toString());
            peers = event.getPeerList();
            myLeader = event.getLeader();
            isLeader = self.equals(myLeader);
        }
    };

    private Handler<GetSuccessorListRequest> handleGetSuccessorListRequest = new Handler<GetSuccessorListRequest>() {
        public void handle(GetSuccessorListRequest event) {
            logger.debug("GET_SUCC_LIST_REQ");

            GetSuccessorListResponse response = new GetSuccessorListResponse(
                    successorList, event.getRequestState(), self,
                    event.getDRSMSource());
            // reply with successorList
            trigger(response, network);
        }
    };

    private Handler<GetSuccessorListResponse> handleGetSuccessorListResponse = new Handler<GetSuccessorListResponse>() {
        public void handle(GetSuccessorListResponse event) {
            // ignore message if it is not sent by my current successor
            if (!successorList.getSuccessor().equals(event.getDRSMSource())) {
                return;
            }

            logger.debug(
                    "GET_SUCC_LIST_RESP from {} my={} got={}",
                    new Object[]{event.getDRSMSource(),
                            successorList.getSuccessors(),
                            event.getSuccessorList().getSuccessors()});

            successorList.updateSuccessorList(event.getSuccessorList());

            // trigger newSuccessor
            DRSMNewSuccessors newSuccessorList = new DRSMNewSuccessors(self,
                    successorList.getSuccessorListView());
            trigger(newSuccessorList, ps);
        }
    };

    private Handler<StabilizeTimeout> handleStabilizeTimeout = new Handler<StabilizeTimeout>() {
        public void handle(StabilizeTimeout event) {
            logger.debug("STABILIZATION s={}, p={}",
                    successorList.getSuccessor(), predecessors);

            GetPredecessorRequest request = new GetPredecessorRequest(self,
                    successorList.getSuccessor());
            // send get predecessors request
            trigger(request, network);

            logger.debug("SENT GET_PRED_REQ to {}",
                    successorList.getSuccessor());
        }
    };

    private Handler<GetPredecessorRequest> handleGetPredecessorRequest = new Handler<GetPredecessorRequest>() {
        public void handle(GetPredecessorRequest event) {
            logger.debug("GET_PRED_REQ from {}", event.getDRSMSource()
                    .getKey());

            GetPredecessorResponse response = new GetPredecessorResponse(
                    predecessors, self, event.getDRSMSource());
            // reply with predecessors
            trigger(response, network);
        }
    };

    private Handler<GetPredecessorResponse> handleGetPredecessorResponse = new Handler<GetPredecessorResponse>() {
        public void handle(GetPredecessorResponse event) {
            // FIXME
            // ignore message if it is not sent by my current successor
            // if (!successorList.getSuccessor().equals(event.getChordSource()))
            // {
            // return;
            // }
            //
            // List<DRSMAddress> predecessorOfMySuccessor =
            // event.getPredecessor();
            //
            // logger.debug("GET_PRED_RESP {} from {} ",
            // predecessorOfMySuccessor,
            // event.getChordSource());
            //
            // if (predecessorOfMySuccessor != null) {
            // if (predecessorOfMySuccessor.getKey().belongsTo(self.getKey(),
            // successorList.getSuccessor().getKey(),
            // IntervalBounds.OPEN_OPEN, ringSize)) {
            //
            // DRSMAddress oldSuccessor = successorList.getSuccessor();
            // DRSMAddress newSuccessor = predecessorOfMySuccessor;
            //
            // successorList.setSuccessor(predecessorOfMySuccessor);
            //
            // neighborReplaced(oldSuccessor, newSuccessor);
            //
            // // trigger newSuccessor
            // NewSuccessorList newSuccessorList = new NewSuccessorList(
            // self, successorList.getSuccessorListView());
            // trigger(newSuccessorList, ps);
            // }
            // }
            //
            // // trigger GetSuccessorList
            // GetSuccessorListRequest request = new GetSuccessorListRequest(
            // RequestState.STABILIZE, self, successorList.getSuccessor());
            // trigger(request, network);
            //
            // Notify notify = new Notify(self, successorList.getSuccessor());
            // // send notify
            // trigger(notify, network);
        }
    };

    private Handler<Notify> handleNotify = new Handler<Notify>() {
        public void handle(Notify event) {
            // FIXME
            // logger.debug("NOTIFY from {}", event.getChordSource());
            //
            // DRSMAddress potentialNewPredecessor = event.getFromPeer();
            //
            // if (predecessors == null
            // || potentialNewPredecessor.getKey().belongsTo(
            // predecessors.getKey(), self.getKey(),
            // IntervalBounds.OPEN_OPEN, ringSize)) {
            //
            // DRSMAddress oldPredecessor = predecessors;
            //
            // predecessors = potentialNewPredecessor;
            //
            // neighborReplaced(oldPredecessor, predecessors);
            //
            // // trigger newPredecessor
            // DRSMNewPredecessor newPredecessor = new DRSMNewPredecessor(self,
            // predecessors);
            // trigger(newPredecessor, ps);
            // }
        }
    };

    private Handler<PeerFailureSuspicion> handlePeerFailureSuspicion = new Handler<PeerFailureSuspicion>() {
        public void handle(PeerFailureSuspicion event) {
            logger.debug("FAILURE_SUSPICION");

            if (event.getSuspicionStatus().equals(SuspicionStatus.SUSPECTED)) {
                // peer is suspected
                DRSMAddress suspectedPeer = neighborPeers.get(event
                        .getPeerAddress());

                if (suspectedPeer == null
                        || !fdRequests.containsKey(suspectedPeer
                        .getPeerAddress())) {
                    // due to component concurrency it is possible that the FD
                    // component sent us a suspicion event right after we sent
                    // it a stop monitor request
                    return;
                }

                // FIXME incomaprable equals
                if (suspectedPeer.equals(predecessors)) {
                    // predecessors suspected
                    predecessors = null;

                    // trigger newPredecessor
                    // FIXME only take out suspected peer from the predecessors
                    // list
                    DRSMNewPredecessors newPredecessor = new DRSMNewPredecessors(
                            self, predecessors);
                    trigger(newPredecessor, ps);
                    neighborRemoved(suspectedPeer);
                }
                if (successorList.getSuccessors().contains(suspectedPeer)) {
                    // successor suspected
                    DRSMAddress oldSuccessor = successorList.getSuccessor();

                    successorList.successorFailed(suspectedPeer);

                    DRSMAddress newSuccessor = successorList.getSuccessor();

                    neighborReplaced(oldSuccessor, newSuccessor);

                    // trigger newSuccessor
                    DRSMNewSuccessors newSuccessorList = new DRSMNewSuccessors(
                            self, successorList.getSuccessorListView());
                    trigger(newSuccessorList, ps);
                }
            } else {
                // peer is alive again
            }
        }
    };

    private void neighborReplaced(DRSMAddress oldPeer, DRSMAddress newPeer) {
        if (oldPeer == null) {
            if (newPeer == null) {
                // both null
                return;
            } else {
                // old is null
                neighborAdded(newPeer);
            }
        } else {
            if (newPeer == null) {
                // new is null
                neighborRemoved(oldPeer);
            } else {
                // none is null
                if (!newPeer.equals(oldPeer)) {
                    // different peers
                    neighborAdded(newPeer);
                    neighborRemoved(oldPeer);
                }
            }
        }
    }

    private void neighborAdded(DRSMAddress peer) {
        if (!peer.equals(self)) {
            Address addr = peer.getPeerAddress();
            if (!neighborPeers.containsKey(addr)) {
                // start failure detection on new neighbor
                StartProbingPeer spp = new StartProbingPeer(addr, peer);
                neighborPeers.put(addr, peer);
                neighborCounters.put(addr, 1);
                fdRequests.put(addr, spp.getRequestId());
                trigger(spp, epfd);
            } else {
                // already a neighbor
                neighborCounters.put(addr, 1 + neighborCounters.get(addr));
            }
            //
            // System.err.println(neighborCounters.get(addr) +
            // " NEIGHBOR_ADDED "
            // + peer + " AT " + self + " MY=" + neighborPeers.values()
            // + " C=" + neighborCounters + " called from "
            // + Thread.currentThread().getStackTrace()[3] + " and from "
            // + Thread.currentThread().getStackTrace()[4]);
        }
    }

    private void neighborRemoved(DRSMAddress peer) {
        if (!peer.equals(self)) {
            Address addr = peer.getPeerAddress();
            if (neighborPeers.containsKey(addr)) {
                int count = neighborCounters.get(addr);
                count--;
                if (count == 0) {
                    // stop failure detection on neighbor
                    neighborPeers.remove(addr);
                    neighborCounters.remove(addr);
                    trigger(new StopProbingPeer(addr, fdRequests.get(addr)),
                            epfd);
                    fdRequests.remove(addr);
                } else {
                    // still a neighbor
                    neighborCounters.put(addr, count);
                }
            }
            //
            // System.err.println(neighborCounters.get(addr)
            // + " NEIGHBOR_REMOVED " + peer + " AT " + self + " MY="
            // + neighborPeers.values() + " C=" + neighborCounters
            // + " called from "
            // + Thread.currentThread().getStackTrace()[3] + " and from "
            // + Thread.currentThread().getStackTrace()[4]);
        }
    }

    private void updateMyKey(NumericRingKey key) {
        self.setKey(key);
    }
}
